package transformationPractice

import org.apache.spark.{SparkConf, SparkContext}

object TransformationOperationScala {

  def init(name:String)={
    val conf = new SparkConf()
      .setAppName(name)
      .setMaster("local")

    new SparkContext(conf)

  }

  def main(args: Array[String]): Unit = {
//    map()
//    filter()
//    flatMap()
//    groupByKey()
    reduceByKey()
  }

  /**
   * map 操作
   */
  def map(): Unit ={
    val sc = init("map")

    val numbers = Array(1,2,3,4,5);

    val numberRDD = sc.parallelize(numbers, 1)

    val multipleNumberRDD = numberRDD.map(_ * 2)

    multipleNumberRDD.foreach(println(_))
  }

  /**
   * fliter 操作实例
   */

  def filter(): Unit ={
    val sc = init("filter")
    val numbers = Array(1,2,3,4,5,6,7,8,9,10)

    val numberRDD = sc.parallelize(numbers, 1)

    val filterRDD = numberRDD.filter(_%2 == 0)

    filterRDD.foreach(println(_))


  }

  def flatMap(): Unit ={
    val sc = init("flatMap")
    val lineArray = Array("hello you", "hello me", "hello world")
    val lines = sc.parallelize(lineArray, 1)
    //
    val words = lines.flatMap(x=>x.split(" "))

    words.foreach(println(_));
    /**
     * 结果：
     * hello
     * you
     * hello
     * me
     * hello
     * world
     */
  }



  /**
   * groupByKey
   */
  def groupByKey(): Unit ={
    val sc = init("groupByKey")

    val scoreList = Array(
      Tuple2("class1", 80),
      Tuple2[String, Integer]("class2", 90),
      Tuple2[String, Integer]("class1", 72),
      Tuple2[String, Integer]("class2", 77),
      Tuple2[String, Integer]("class1", 93),
      Tuple2[String, Integer]("class2", 124)
    )

    val scores = sc.parallelize(scoreList,1)
    val groupedScores = scores.groupByKey()
    groupedScores.foreach(score=>{
      println(score._1);
      score._2.foreach(println(_))
    })
  }


  /**
   * reduceByKey
   *
   */

  def reduceByKey(): Unit ={
    val sc = init("reduceByKey")

    val scoreList = List(
      ("class1", 80),
      ("class2", 90),
      ("class1", 72),
      ("class2", 77),
      ("class1", 93),
      ("class2", 124)
    )

    val scores = sc.parallelize(scoreList,1)

    val totalScores = scores.reduceByKey(_+_)

    totalScores.foreach(println(_))
    totalScores.foreach(x=>println(x._1+" : "+x._2))
  }


}
